package com.supreme.commonrocketmq.consumer;

import com.alibaba.fastjson.JSONObject;
import com.supreme.common.entityRocketmq.MQLogger;
import com.supreme.common.entityRocketmq.MQOrder;
import com.supreme.commonrocketmq.config.props.RocketmqProps;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;
import java.util.function.Consumer;

@Slf4j
@Component
public class RocketmqConsumer {

    @Resource
    private RocketmqProps rocketmqProps;

    @Resource
    private ApplicationContext applicationContext;

    /**
     * 创建基本测试消费端
     * @param filterType: 过滤类型
     */
    private void createMqGroupConsumer(Integer filterType) {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
        defaultMQPushConsumer.setNamesrvAddr(rocketmqProps.getProducerNameSrv());
        defaultMQPushConsumer.setConsumerGroup(rocketmqProps.getExample().getGroup());
        try {
            switch (filterType) {
                case 1:
                    defaultMQPushConsumer.subscribe(rocketmqProps.getExample().getName(), "*");
                    break;

                case 2:
                    defaultMQPushConsumer.subscribe(rocketmqProps.getExample().getName(), "hello || world");
                    break;

                case 3:
                    defaultMQPushConsumer.subscribe(rocketmqProps.getExample().getName(), MessageSelector.byTag("hello || world"));
                    break;

//                case 4:
//                    defaultMQPushConsumer.subscribe(rocketmqProps.getExample().getName(), MessageSelector.bySql("condition >=0 and condition <= 3"));
//                    break;
            }
            defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExtList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    messageExtList.forEach(new Consumer<MessageExt>() {
                        @Override
                        public void accept(MessageExt messageExt) {
                            String topic = messageExt.getTopic();
                            String tags = messageExt.getTags();
                            String keys = messageExt.getKeys();
                            System.out.printf("topic: %s tags: %s keys: %s\n", topic, tags, keys);

                            byte[] bodyBytes = messageExt.getBody();
                            JSONObject bodyJson = JSONObject.parseObject(bodyBytes, JSONObject.class);
                            System.out.println("bodyJson: " + bodyJson);
                        }
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            defaultMQPushConsumer.start();
        } catch (MQClientException e) {
            log.error(e.getMessage(), e);
        }
    }

    private void createOrderGroupConsumer() {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
        defaultMQPushConsumer.setNamesrvAddr(rocketmqProps.getProducerNameSrv());
        defaultMQPushConsumer.setConsumerGroup(rocketmqProps.getOrder().getGroup());
        try {
            defaultMQPushConsumer.subscribe(rocketmqProps.getOrder().getName(), "*");
            defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExtList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    messageExtList.forEach(new Consumer<MessageExt>() {
                        @Override
                        public void accept(MessageExt messageExt) {
                            byte[] bodyBytes = messageExt.getBody();
                            MQOrder order = JSONObject.parseObject(bodyBytes, MQOrder.class);
                            log.info("order: {}", order);
                        }
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            defaultMQPushConsumer.start();
        } catch (MQClientException e) {
            log.error(e.getMessage(), e);
        }
    }

    private void createLoggerGroupConsumer() {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
        defaultMQPushConsumer.setNamesrvAddr(rocketmqProps.getProducerNameSrv());
        defaultMQPushConsumer.setConsumerGroup(rocketmqProps.getLogger().getGroup());
        try {
            defaultMQPushConsumer.subscribe(rocketmqProps.getLogger().getName(), "*");
            defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExtList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    messageExtList.forEach(new Consumer<MessageExt>() {
                        @Override
                        public void accept(MessageExt messageExt) {
                            byte[] bodyBytes = messageExt.getBody();
                            MQLogger logger = JSONObject.parseObject(bodyBytes, MQLogger.class);
                            log.info("logger: {}", logger);
                        }
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            defaultMQPushConsumer.start();
        } catch (MQClientException e) {
            log.error(e.getMessage(), e);
        }
    }

    private void createMysqlBinLogConsumer() {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
        defaultMQPushConsumer.setNamesrvAddr(rocketmqProps.getProducerNameSrv());
        defaultMQPushConsumer.setConsumerGroup(rocketmqProps.getMysqlBinlog().getGroup());
        try {
            defaultMQPushConsumer.subscribe(rocketmqProps.getMysqlBinlog().getName(), "*");
            defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExtList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    messageExtList.forEach(new Consumer<MessageExt>() {
                        @Override
                        public void accept(MessageExt messageExt) {
                            byte[] bodyBytes = messageExt.getBody();
                            JSONObject bodyJson = JSONObject.parseObject(bodyBytes, JSONObject.class);
                            System.out.println("bodyJson = " + bodyJson);

                            String database = bodyJson.getString("database");
                            String table = bodyJson.getString("table");
                            String type = bodyJson.getString("type");
                            /*if (StringUtils.equals(database, "supreme_dev") && StringUtils.equals(table, "test")) {
                                ESTest[] dataList = bodyJson.getObject("data", ESTest[].class);
                                if (dataList.length > 0) {
                                    ESTest esTest = dataList[0];
                                    try {
                                        String documentId = String.valueOf(esTest.getId());
                                        if (StringUtils.equals(type, "INSERT")) {
                                            elasticsearchService.createDocument("mysql_test_index", esTest, documentId);
                                        } else if (StringUtils.equals(type, "UPDATE")) {
                                            elasticsearchService.updateDocument("mysql_test_index", documentId, esTest);
                                        }
                                    } catch (IOException e) {
                                        throw new RuntimeException(e);
                                    }
                                }
                            }*/
                        }
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            defaultMQPushConsumer.start();
        } catch (MQClientException e) {
            log.error(e.getMessage(), e);
        }
    }

    @PostConstruct
    private void createRocketmqConsumer() {
        createMqGroupConsumer(4);
        createOrderGroupConsumer();
        createLoggerGroupConsumer();
        createMysqlBinLogConsumer();
    }
}
